package lab3;

import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Scanner;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;



public class GraphBuilder {
	public static class Map extends Mapper<LongWritable,Text,Text,Text>{
		
		
		public void map(LongWritable key,Text value,Context context)
				throws  IOException, InterruptedException{
			
			// 将一行数据读入， 并将字符串进行分割
			 String line = value.toString();  
			 String[]tuple = line.split("\t");
	         
			
			 // 当前网页
			 String page = tuple[0]; 
			 // 包含当前初始的 pr 值 1 ， link的链表， 以空格分割
			 String pr_link = "1";
			 // 将连接加入链表
			 pr_link = pr_link + "\t" + tuple[1];  
			 
			 Text k = new Text(page);
			 Text v = new Text();  
             v.set(pr_link); 
             context.write(k,v);
           
		}
	}
	public static class Reduce extends Reducer<Text,Text,Text,Text> {
		
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			for(Text val : values){
				context.write(key, val);
			}
		}
	}
	
	public static void main(String[] args) throws Exception{
		// TODO Auto-generated method stub
		 Configuration conf = new Configuration();
		 Path path =new Path(args[1]);
		  FileSystem fileSystem = path.getFileSystem(conf);
	      if (fileSystem.exists(new Path(args[1]))) { 
	      	fileSystem.delete(new Path(args[1]),true); 
	      }
		  Job job = new Job(conf, "Graph Builder ");
		  job.setJarByClass(GraphBuilder.class);
		  job.setMapperClass(Map.class);
		  job.setReducerClass(Reduce.class);
		  job.setOutputKeyClass(Text.class);
		  job.setOutputValueClass(Text.class);
		  FileInputFormat.addInputPath(job, new Path(args[0]));
		  FileOutputFormat.setOutputPath(job, new Path(args[1]));
		 job.waitForCompletion(true);

	}

}
